package com.central.service

import java.sql.{DriverManager, Timestamp}
import java.util
import java.util.{Calendar, Date, Properties, UUID}

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.central.bean
import com.central.bean.WifiEntity
import com.central.controller.{OnePointTrailConditionEntity, TrailConditionEntity}
import com.central.utils.Relationship.{AND, NOT, OR, Relationship}
import com.central.utils.{ConfigUtil, LoadESData, ParseTime}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.elasticsearch.spark.rdd.EsSpark

import scala.reflect.runtime.universe.TypeTag
import scala.reflect.ClassTag

class TrailService {

  val TABLE_TRAIL_RESULT = ConfigUtil.getProperties().getProperty("table.trail.result")
  val TABLE_TRAIL_DETAIL = ConfigUtil.getProperties().getProperty("table.trail.detail")
  val JDBC_URL = ConfigUtil.getProperties().getProperty("mysql.url")

  val TRAIL_INDEX: String = ConfigUtil.getProperties().getProperty("trail.es.index")
  val TABLE_TRAIL_STATUS = ConfigUtil.getProperties().getProperty("table.trail.status")
  val USER_NAME = ConfigUtil.getProperties().getProperty("mysql.username")
  val PASSWORD = ConfigUtil.getProperties().getProperty("mysql.password")

  def trailAnalysis(spark: SparkSession, trailContext: TrailConditionEntity) = {
    val jobId = trailContext.jobId
    val threshold = trailContext.threshold
    val hasliferow = trailContext.hasliferow
    val fuzzyway = trailContext.fuzzyway
    val activity = trailContext.activity
    val listRdd: List[RDD[TrailResultWithDetail]] = trailContext
      .conditions
      .map(condition => {
        val esRdd = trailUnionNode(spark, condition, jobId + "")
        //生活规律分析
        esRdd
        //          .distinct()
      })
          .filter(x => {
            x != null
          })
    val joinRdd: RDD[TrailResultWithDetail] = listRdd.foldLeft(spark.sparkContext.emptyRDD[TrailResultWithDetail]) {
      case (prevRdd, oneNodeRdd) => {
        if (prevRdd.isEmpty() && !oneNodeRdd.isEmpty()) {
          oneNodeRdd
        }
        else if (!oneNodeRdd.isEmpty()) {
          mergeRddsWith(OR, prevRdd, oneNodeRdd)(entity => entity.idNumber, (one, another) => one.mergeTrail(another))
        }
        else
          spark.sparkContext.emptyRDD[TrailResultWithDetail]
      }
    }

    val hasliferowRdd = if (hasliferow==1) {
      val recentRdd = LoadESData.loadRecentEs(spark, TRAIL_INDEX, activity)
      joinRdd.map(data=>{
        (data.idNumber,data)
      })
        .join(recentRdd)
        .map(x=>{
          x._2._1
        })
    }
    else{
      joinRdd
    }
    //模糊处理
    println(joinRdd.count())
    println("==========模糊处理==========")
    val fuzzywayRdd = if (fuzzyway != 0) {
      val alldevicecodes: List[String] = trailContext.conditions.flatMap(x => {
        x.deviceids
      }).distinct
      val details: Array[TrailResultWithDetail] = hasliferowRdd.collect()
        .map(data => {
          //只对满足条件数小于阈值且小于阈值不超过2的
          if (data.conditionCount < threshold && (threshold - data.conditionCount) < 2) {
            println(data.conditionCount+" ==============conditioncount")
            val codetype = data.dataType
            val tagcode = data.idNumber
            val details: util.ArrayList[TrailResultDetail] = data.details
            val device_codes: Array[String] = details.toArray.map(x => {
              x.asInstanceOf[TrailResultDetail].devicecode + ""
            })

            val diffDevice: List[String] = alldevicecodes.diff(device_codes)

            //通过关联关系号码，找出关联号码经常出现的点位信息
            if (fuzzyway == 1) {
              println("==========trial type1==========")
              var relat_code = findRelatDevice(spark, TRAIL_INDEX, tagcode)
              if (relat_code != null) {
                if (diffDevice.intersect(relat_code).size > 0) {

                  val count = data.conditionCount + 1
                  TrailResultWithDetail(data.jobId, data.idNumber, data.dataType, data.count, data.details, count, data.resultid, data.createtime, data.address, data.phone)
                }
              }
            }
            else if (fuzzyway == 2) {
              println("==========trial type2==========")
              var relat_code = findTrailDevice(spark, TRAIL_INDEX, tagcode, codetype)
              if (relat_code != null) {
                if (diffDevice.intersect(relat_code).size > 0) {
                  val count = data.conditionCount + 1
                  TrailResultWithDetail(data.jobId, data.idNumber, data.dataType, data.count, data.details, count, data.resultid, data.createtime, data.address, data.phone)
                }
              }
            }
            else {
              println("==========trial type3==========")
              var relat_code = findHotDevice(spark, TRAIL_INDEX, tagcode, codetype)
              if (relat_code != null) {
                if (diffDevice.intersect(relat_code).size > 0) {
                  val count = data.conditionCount + 1
                  TrailResultWithDetail(data.jobId, data.idNumber, data.dataType, data.count, data.details, count, data.resultid, data.createtime, data.address, data.phone)
                }
              }
            }
          }
          data

        })
      spark.sparkContext.parallelize(details)
    }
    else hasliferowRdd

    println(fuzzywayRdd.count())
    val trailResultRdd: RDD[TrailResultWithDetail] = fuzzywayRdd.filter(result => result.conditionCount >= threshold)

    //保存到数据库当中
    val finalResult: RDD[TrailResultWithDetail] = if (trailResultRdd.isEmpty()) spark.sparkContext
      .parallelize(List(TrailResultWithDetail(trailContext.jobId + "", "0", "None", 0, null, 0, null, null, null, null))) else trailResultRdd
    val properties = new Properties()
    properties.put("user", "root")
    properties.put("password", "root")
    val trailResult = finalResult
      .map(x => {

        val jobId: String = x.jobId
        val code = x.idNumber
        val dataType = x.dataType
        val count = x.count
        val conditionCount = x.conditionCount
        val phone = x.phone
        TrailResult(jobId.toInt, code, dataType, count, ParseTime.LongToDate(System.currentTimeMillis().toString), conditionCount, x.resultid, x.address, phone)
        //        TrailResult(1121, code, dataType, 1, ParseTime.LongToDate(System.currentTimeMillis().toString), 1, x.resultid, x.location, "123")
      })
      .cache()
    import spark.implicits._
    println(trailResult.count() + ">>>>>>>>>")
    trailResult
      .filter(x => {
        !(x.codetype.equals("None"))
      })
      .
        toDF()
      .write
      .mode(SaveMode.Append)
      .jdbc(JDBC_URL, TABLE_TRAIL_RESULT, properties)
    finalResult
      .filter(x => {
        !(x.dataType.equals("None"))
      })
      .flatMap(x => {
        x.details.toArray()
      })
      .map(x => {
        x.asInstanceOf[TrailResultDetail]
      })
      .filter(x => {
        !("None".equals(x.resultid)) || !("None".equals(x.codetype))
      })
      .toDS().write.mode(SaveMode.Append).jdbc(JDBC_URL, TABLE_TRAIL_DETAIL, properties)

    updateStatus(jobId + "")
  }

  def updateStatus(jobId: String) = {
    val table = TABLE_TRAIL_STATUS
    try {
      Class.forName("com.mysql.cj.jdbc.Driver")
    } catch {
      case e =>
        e.printStackTrace()
    }
    val connection = DriverManager.getConnection(JDBC_URL, USER_NAME, PASSWORD)
    val statement = connection.prepareStatement(
      s"update  $table set status = ?, successtime=? where id = ?  "
    )
    statement.setInt(1, 2)
    statement.setString(2, ParseTime.LongToDate(System.currentTimeMillis().toString))
    statement.setInt(3, jobId.toInt)
    statement.executeUpdate()
    statement.close()
    connection.close()
  }


  def trailUnionNode(spark: SparkSession, condition: OnePointTrailConditionEntity, jobId: String): RDD[TrailResultWithDetail] = {
    val continue = condition.continue
    val begintime = ParseTime.DateToLong(condition.begintime)
    val endtime = begintime + continue * 60 * 1000

    val deviceids = condition.deviceids
    import scala.collection.JavaConverters._

    val esRdd: RDD[(String, collection.Map[String, AnyRef])] = LoadESData.getMutliDeviceData(spark, TRAIL_INDEX, deviceids.asJava, ParseTime.LongToDate(begintime + ""), ParseTime.LongToDate(endtime + ""))

    var trailRDD: RDD[WifiEntity] = null;
    println(esRdd.count() + "===========es count")
    if (esRdd.count() > 0) {

      if (trailRDD == null) {
        trailRDD = esRdd.map(e => {
          if (e._2.get("mac").get != None) //mac数据的封装
            WifiEntity(
              e._1, e._2.get("mac").get + "",
              e._2.get("mac").get + "",
              "mac",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),

              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").get + "",
                          e._2.get("location").get+"",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),


              e._2.get("phone").get+"",
              "",
              e._2.get("picpath").get+""
            )
          else if (e._2.get("imsi").get != None)
            WifiEntity(
              e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "imsi",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").get + "",
                          e._2.get("location").get+"",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").get+"", "",
              e._2.get("picpath").get+""
            )
          else
            WifiEntity(
              e._1, e._2.get("imei").get + "", e._2.get("imei").get + "", "imei",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").get + "",
                          e._2.get("location").get+"",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").get+"", "",
              e._2.get("picpath").get+""
            )
        })
      }
      else {
        trailRDD = trailRDD ++ esRdd.map(e => {
          if (e._2.get("mac").get != None) //mac数据的封装
            WifiEntity(
              e._1, e._2.get("mac").get + "",
              e._2.get("mac").get + "",
              "mac",
              Timestamp.valueOf(e._2.get("captime").get + ""),
              Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),

              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").get + "",
                          e._2.get("location").get+"",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").get+"",
              "",
              e._2.get("picpath").get+""
            )
          else if (e._2.get("imsi").get != None)
            WifiEntity(
              e._1, e._2.get("imsi").get + "", e._2.get("imsi").get + "", "imsi",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").get + "",
                          e._2.get("location").get+"",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),
              e._2.get("phone").get+"", "",
              e._2.get("picpath").get+""
            )
          else
            WifiEntity(
              e._1, e._2.get("imei").get + "", e._2.get("imei").get + "", "imei",
              Timestamp.valueOf(e._2.get("captime").get + ""), Timestamp.valueOf(e._2.get("captime").get + ""),
              //            Timestamp.valueOf(e._2.get("ir_time").get + ""),
              new Date(ParseTime.DateToLong(e._2.get("ir_time").get + "")),
              e._2.get("device_code").get + "",
              e._2.get("device_name").get + "",
              e._2.get("address").get + "",
                          e._2.get("location").get+"",
              e._2.get("captime").get + "",
              java.lang.Double.parseDouble(e._2.get("latitude").get + ""),
              java.lang.Double.parseDouble(e._2.get("longitude").get + ""),

              e._2.get("phone").get+"", "",
              e._2.get("picpath").get+""
            )
        })
      }
    }
    else {
      println("未找到相应的数据")
    }

    //    val value = trailRDD.map(wifiEntity => {
    //      (wifiEntity.idNumber, wifiEntity)
    //    })
    //      .groupByKey()
    //    println("==================value")

    if (trailRDD == null) {
      return null
    }
    else {
      val resultRdd: RDD[TrailResultWithDetail] = trailRDD
        .map(wifiEntity => {
          (wifiEntity.idNumber, wifiEntity)
        })
        .groupByKey()
        .map { case (idNumber, sameEntityList) => {
          val id = UUID.randomUUID() + ""

          val details = sameEntityList.map(x => {
            val resultId = x.id
            val deviceNumber = x.deviceNumber.toInt
            val tagcode = x.idNumber
            val codeType = x.idNumberType
            val deviceName = x.deviceName
            val address = x.address
            //        val location = x.location
            val phoneNumber = x.phoneNumber
            val captime = x.captime
            TrailResultDetail(
              id,
              x.idNumber,
              x.idNumberType,
              x.deviceNumber.toInt,
              deviceName,
              address,
              null, captime, x.latitude, x.longitude)
          }).toList

          val entities = new util.ArrayList[TrailResultDetail]()
          for (i <- 0 to details.size - 1) {
            entities.add(details(i))
          }
          //      ClueResult(
          //        jobId,
          //        sameEntityList.head.showedIdNumber,
          //        sameEntityList.head.idNumberType,
          //        sameEntityList.size,
          //        sameEntityList.map(entity => entity.id).mkString(";")
          //      )
          TrailResultWithDetail(
            jobId,
            sameEntityList.head.showedIdNumber,
            sameEntityList.head.idNumberType,
            sameEntityList.size,
            entities,
            1,
            id,
            sameEntityList.head.createTime,
            sameEntityList.head.address,
            null

          )
        }
        }
      resultRdd
    }
  }

  private[service] def mergeRddsWith[T: ClassTag : TypeTag](relationShip: Relationship, one: RDD[T], another: RDD[T])(key: T => String, merge: (T, T) => T): RDD[T] = {

    val onePair =
      one.mapPartitions(x => {
        x.map(x => {
          (key(x), x)
        })
      })
    val anotherPair =
      another.mapPartitions(x => {
        x.map(x => {
          (key(x), x)
        })
      })


    relationShip match {
      case AND => onePair.fullOuterJoin(anotherPair).map {
        case (key, joinResult) => {
          if (joinResult._1.isDefined && joinResult._2.isDefined)
            merge(joinResult._1.get, joinResult._2.get)
          else
            joinResult._1.getOrElse(joinResult._2.get)
        }
      }
      case OR => onePair.fullOuterJoin(anotherPair).map {
        case (key, joinResult) => {
          if (joinResult._1.isDefined && joinResult._2.isDefined)
            merge(joinResult._1.get, joinResult._2.get)
          else
            joinResult._1.getOrElse(joinResult._2.get)
        }
      }
      case NOT => onePair
        .leftOuterJoin(anotherPair)
        .filter(result => result._2._2.isEmpty)
        .mapPartitions(x => {
          x.map(_._2._1)
        })
      //        .optMap(_._2._1)
    }
  }

  /**
    * 找出与参数code相关联的号码对应的经常出现的设备点
    *
    * @param code
    */
  def findRelatDevice(spark: SparkSession, index: String, code: String) = {

    //找出code关联的号码
    try {
      //      Class.forName("com.mysql.jdbc.Driver")
      Class.forName("com.mysql.cj.jdbc.Driver")
    } catch {
      case e =>
        e
    }
    var relatcode = new util.ArrayList[String]()
    val connection = DriverManager.getConnection(JDBC_URL, USER_NAME, PASSWORD)
    val statement = connection.prepareStatement(s"select tagcode,tagtype from t_wifi_eachother_relation where srccode=$code order by counts desc")
    val resultSet = statement.executeQuery()
    var data = ""
    if (resultSet.next()) {
      data = (resultSet.getString(1) + "_" + resultSet.getString(2))
    }
    resultSet.close()
    statement.close()
    connection.close()

    if (data.length > 0) {
      val codetype = data.split("_")(0)
      val tagcodes = data.split("_")(1)

      val queryPara = new JSONObject()
      val queryBool = new JSONObject()
      val queryCode = new JSONObject()
      if (codetype.equals("imsi")) {
        queryCode.put("imsi", tagcodes)
      }
      else if (codetype.equals("mac")) {
        queryCode.put("mac", tagcodes)
      }
      else {
        queryCode.put("imei", tagcodes)
      }

      val calendar = Calendar.getInstance()
      val date = new Date()
      calendar.setTime(date)
      calendar.add(Calendar.DATE, 0 - 10)
      calendar.set(Calendar.HOUR, 0)
      calendar.set(Calendar.AM_PM, 0)
      calendar.set(Calendar.MINUTE, 0)
      calendar.set(Calendar.SECOND, 0)
      val starttime = ParseTime.sdf.format(calendar.getTime)
      calendar.add(Calendar.DATE, 10)
      val endtime = ParseTime.sdf.format(calendar.getTime)


      val queryTerms = new JSONObject()
      queryTerms.put("term", queryCode)
      val queryFilters = new JSONArray()
      queryFilters.add(queryTerms)
      val queryTime = new JSONObject()
      queryTime.put("gte", starttime)
      queryTime.put("lte", endtime)
      val querycaptime = new JSONObject()
      querycaptime.put("captime", queryTime)
      val queryRange = new JSONObject()
      queryRange.put("range", querycaptime)
      queryFilters.add(queryRange)

      queryBool.put("filter", queryFilters)
      queryPara.put("bool", queryBool)
      val esRdd = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString)
      esRdd.map(x => {
        x._2.get("device_code").get + ""
      })
        .map(x => {
          (x, 1)
        })
        .reduceByKey(_ + _)
        .collect()
        .sortBy(x => {
          x._2
        })
        .reverse
        .map(x => {
          x._1
        })
        .splitAt(3)
        ._1
    }
    else null
  }

  /**
    * 找出code历史经常出现的设备点
    *
    * @param spark
    * @param index
    * @param code
    */
  def findHotDevice(spark: SparkSession, index: String, code: String, codetype: String) = {

    val calendar = Calendar.getInstance()
    val date = new Date()
    calendar.setTime(date)
    calendar.add(Calendar.DATE, 0 - 10)
    calendar.set(Calendar.HOUR, 0)
    calendar.set(Calendar.AM_PM, 0)
    calendar.set(Calendar.MINUTE, 0)
    calendar.set(Calendar.SECOND, 0)
    val starttime = ParseTime.sdf.format(calendar.getTime)
    calendar.add(Calendar.DATE, 10)
    val endtime = ParseTime.sdf.format(calendar.getTime)

    val queryPara = new JSONObject()
    val queryBool = new JSONObject()

    val queryMatch = new JSONObject()
    queryMatch.put(codetype, code)
    val queryMast = new JSONObject()
    queryMast.put("match", queryMatch)
    queryBool.put("must", queryMast)

    val queryTime = new JSONObject()
    queryTime.put("gte", starttime)
    queryTime.put("lte", endtime)
    val querycaptime = new JSONObject()
    querycaptime.put("captime", queryTime)
    val queryRange = new JSONObject()
    queryRange.put("range", querycaptime)
    val queryMasts = new JSONArray()
    queryMasts.add(queryRange)
    val queryBools = new JSONObject()
    queryBools.put("must", queryMasts)
    val queryFilter = new JSONObject()
    queryFilter.put("bool", queryBools)
    queryBool.put("filter", queryFilter)
    queryPara.put("bool", queryBool)
    val esRdd = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString)

    esRdd.map(x => {
      (x._2.get("device_code").get + "", 1)
    })
      .reduceByKey(_ + _)
      .collect()
      .sortBy(x => {
        x._2
      })
      .reverse
      .splitAt(3)
      ._1
      .map(x => {
        x._1
      })
  }

  /**
    * 找出code历史轨迹经常出现的设备点
    *
    * @param spark
    * @param index
    * @param code
    * @param codetype
    */
  def findTrailDevice(spark: SparkSession, index: String, code: String, codetype: String) = {
    val calendar = Calendar.getInstance()
    val date = new Date()
    calendar.setTime(date)
    calendar.add(Calendar.DATE, 0 - 10)
    calendar.set(Calendar.HOUR, 0)
    calendar.set(Calendar.AM_PM, 0)
    calendar.set(Calendar.MINUTE, 0)
    calendar.set(Calendar.SECOND, 0)
    val starttime = ParseTime.sdf.format(calendar.getTime)
    calendar.add(Calendar.DATE, 10)
    val endtime = ParseTime.sdf.format(calendar.getTime)

    val queryPara = new JSONObject()
    val queryBool = new JSONObject()

    val queryMatch = new JSONObject()
    queryMatch.put(codetype, code)
    val queryMast = new JSONObject()
    queryMast.put("match", queryMatch)
    queryBool.put("must", queryMast)

    val queryTime = new JSONObject()
    queryTime.put("gte", starttime)
    queryTime.put("lte", endtime)
    val querycaptime = new JSONObject()
    querycaptime.put("captime", queryTime)
    val queryRange = new JSONObject()
    queryRange.put("range", querycaptime)
    val queryMasts = new JSONArray()
    queryMasts.add(queryRange)
    val queryBools = new JSONObject()
    queryBools.put("must", queryMasts)
    val queryFilter = new JSONObject()
    queryFilter.put("bool", queryBools)
    queryBool.put("filter", queryFilter)
    queryPara.put("bool", queryBool)
    val esRdd = EsSpark.esRDD(spark.sparkContext, index, queryPara.toJSONString)

    esRdd.map(x => {
      (x._2.get("device_code").get + "", 1)
    })
      .reduceByKey(_ + _) //device,count
      .map(x => {
      x.swap
    })
      .groupByKey() //count,devices
      .collect()
      .sortBy(_._1)
      .reverse
      .splitAt(3) //点位count出现次数拍前三的
      ._1
      .sortBy(tump => {
        //按出现点位数相同的点位数据集个数进行排序
        tump._2.size
      })
      .reverse
      .head //得到出现点位数相同的点位数据集个数最多的
      ._2
      .toArray
  }

}

case class TrailResultDetail(
                              //                              id: String,
                              var resultid: String,
                              tagCode: String,
                              codetype: String,
                              devicecode: Integer,
                              deviceName: String,
                              address: String,
                              phone: String,
                              captime: String,
                              latitude: java.lang.Double,
                              longtitude: java.lang.Double
                            ) {
  def setResultId(string: String) = {
    resultid = string
  }
}

case class TrailResult(
                        jobid: Integer,
                        tagcode: String,
                        codetype: String,
                        count: Integer,
                        createtime: String,
                        hitcount: Integer,

                        var resultid: String,
                        address: String,
                        phone: String
                      ) {
  def setResultId(string: String) = {
    resultid = string
  }
}

case class TrailResultWithDetail(jobId: String,
                                 idNumber: String,
                                 dataType: String,
                                 count: Int,
                                 details: util.ArrayList[TrailResultDetail],
                                 conditionCount: Int = 1,

                                 resultid: String,
                                 createtime: Date,
                                 address: String,
                                 phone: String
                                ) {
  def mergeTrail(another: TrailResultWithDetail): TrailResultWithDetail = {
    if (idNumber.equals(another.idNumber)) {

      for (i <- 0 to another.details.size() - 1) {
        //进行merge时，需要注意resultId，由于每个
        if (conditionCount >= 1) {
          val preId = details.get(0).resultid
          val detail = another.details.get(i)
          detail.setResultId(preId)
          details.add(detail)
        }
        else {
          details.add(another.details.get(i))
        }
      }
      TrailResultWithDetail(jobId, idNumber, dataType, count + another.count, details, conditionCount + 1, resultid, createtime, address, phone)
    }
    else
      this
  }

  def mergeTrailWithPlusingConditionCount(another: TrailResultWithDetail): TrailResultWithDetail = {
    if (idNumber.equals(another.idNumber)) {
      for (i <- 0 to another.details.size() - 1) {
        if (conditionCount >= 1) {
          val preId = details.get(0).resultid
          val detail = another.details.get(i)
          detail.setResultId(preId)
          details.add(detail)
        }
        else {
          details.add(another.details.get(i))
        }
      }
      TrailResultWithDetail(jobId, idNumber, dataType, count + another.count, details, conditionCount + 1, resultid, createtime, address, phone)
    }
    else
      this
  }
}
